package com.zhang.spark_2.com.zhang.core.req

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * @title: 热门商品Top10
 * @author: zhang
 * @date: 2022/2/16 17:16 
 */
object Spark01_req_pro_max {

  def main(args: Array[String]): Unit = {
    //TODO 获取环境
    val conf = new SparkConf().setMaster("local").setAppName("WordCount")
    val sc = new SparkContext(conf);
    val userVisit: RDD[String] = sc.textFile("data/user_visit_action.txt")
    userVisit.cache()
    // (品类，点击数量）

    val flatRDD: RDD[(String, (Int, Int, Int))] = userVisit.flatMap(
      line => {
        val datas = line.split("_")
        if (datas(6) != "-1") {
          List((datas(6), (1, 0, 0)))
        } else if (datas(8) != "null") {
          val ids: Array[String] = datas(8).split(",")
          ids.map((_, (0, 1, 0)))
        } else if (datas(10) != "null") {
          val ids: Array[String] = datas(10).split(",")
          ids.map((_, (0, 0, 1)))
        } else {
          Nil
        }
      }
    )

    val analysisRDD: RDD[(String, (Int, Int, Int))] = flatRDD.reduceByKey((t1, t2) => (t1._1 + t2._1, t1._2 + t2._2, t1._3 + t2._3))

    analysisRDD.sortBy(_._2, false).take(10).foreach(println)


    sc.stop()
  }
}
